package cn.wangjie.spark.topkey

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Spark实现词频统计：WordCount，对词频count进行降序排序，获取Top3单词
 */
object SparkTopKey {

  def main(args: Array[String]): Unit = {
		
		// TODO: 1. 构建SparkContext上下为实例对象
		val sc: SparkContext = {
			// 1.a 创建SparkConf对象，设置应用属性
			val sparkConf = new SparkConf()
    			.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
    			.setMaster("local[2]")
			// 1.b 传递SparkConf对象，创建上下文实例
			SparkContext.getOrCreate(sparkConf)
		}
		
		// TODO：2. 读取数据，封装为RDD
		val inputRDD: RDD[String] = sc.textFile("/datas/wordcount.data")
		
		// TODO: 3. 调用RDD函数，处理分析数据
		val wordsRDD: RDD[String] = inputRDD.flatMap(line => line.split("\\s+"))
		val tuplesRDD: RDD[(String, Int)] = wordsRDD.map(word => (word, 1))
		val resultRDD: RDD[(String, Int)] = tuplesRDD.reduceByKey((tmp, item) => tmp + item)
		
		// TODO: 4. 将结果RDD存储或打印
		resultRDD.foreach(println)
		
		println("======================== sortByKey ================================")
		
		/**
		 * 对词频count进行降序排序，使用RDD中排序函数完成：
		 * -1. 数组Array或者列表List中排序函数：
		 *      sorted: 按照升序，自然
		 *      sortBy：指定排序字段及升序还是降序
		 *      sortWith: 通过两个值比较确定升序还是降序
		 * -2. 在RDD中提供如何排序函数
		 *      sortByKey: 按照Key进行排序
		 *      sortBy: 指定排序字段
		 *      top: 直接排序获取，但是不建议使用
		 */
		/*
		def sortByKey(
			ascending: Boolean = true,
			numPartitions: Int = self.partitions.length
		): RDD[(K, V)]
		 */
		resultRDD
			.map(tuple => tuple.swap)
			.sortByKey(ascending = false)
    		.take(3)
			.foreach(item => println(item))
		
		println("======================== sortBy ================================")
		
		/*
		  def sortBy[K](
		      f: (T) => K,
		      ascending: Boolean = true,
		      numPartitions: Int = this.partitions.length
		   )
		   // 隐式参数，可以先不问，
		   (implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T]
		 */
		resultRDD
			.sortBy(
				tuple => tuple._2, // 按照count排序
				ascending = false
			)
			.take(3)
			.foreach(item => println(item))
		
		println("======================== top ================================")
		/*
			def top(num: Int)(implicit ord: Ordering[T]): Array[T]
		 */
		resultRDD
			.map(tuple => tuple.swap)
			.top(3)
			.foreach(item => println(item))
		
		Thread.sleep(10000000)
		
		// TODO: 5. 应用运行结束，关闭资源
		sc.stop()
	}
	
}
